package com.github.yoojia.mqtt;

import com.parkingwang.lang.data.Pair;
import com.parkingwang.lang.kit.CastKit;
import com.parkingwang.lang.kit.StringKit;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Yoojia Chen (yoojiachen@gmail.com)
 * @since 1.0.0
 */
final class Subscriber {

    public static final Logger LOGGER = LoggerFactory.getLogger(Subscriber.class);

    private final Map<Long, Pair<LocalFilter, MessageHandler>> mArgsMap = new ConcurrentHashMap<>();

    private final MQTTSocket mSocket;
    private final SnowflakeId mRegSerialId;
    private final MqttClient mClient;
    private final String mTopicFilter;

    public Subscriber(MQTTSocket socket, SnowflakeId snowflakeId, MqttClient client, String topicFilter) {
        mSocket = socket;
        mRegSerialId = snowflakeId;
        mClient = client;
        mTopicFilter = topicFilter;
    }

    public String getTopicFilter() {
        return mTopicFilter;
    }

    public void subscribe() throws MqttException {
        LOGGER.info("Subscribing with filter topic: {}", mTopicFilter);
        mClient.subscribe(mTopicFilter, new IMqttMessageListener() {
            @Override public void messageArrived(String topic, MqttMessage mqttMsg) throws Exception {
                // 接收到消息的Topic规则为：
                // /${string:service-name}/${string:target-id}/${string:pattern-type}/${long:request-id}/${string:senderNodeId-id}/${string:tag}
                final String[] segments = StringKit.split(topic, "/");
                final String serviceName = segments[0];
                final String targetId = segments[1];
                final String type = segments[2];
                final long messageId = CastKit.castLong(segments[3], 0L);
                final String endpointId = segments[4];
                final String tag = segments.length >= 6 ? segments[5] : "";
                final Message receivedMessage = new Message.Builder()
                        .domain(serviceName)
                        .targetNodeId(targetId)
                        .type(type)
                        .messageId(messageId)
                        .senderNodeId(endpointId)
                        .tag(tag)
                        .payload(mqttMsg.getPayload())
                        .qos(mqttMsg.getQos())
                        .retained(mqttMsg.isRetained())
                        .build();
                if (LOGGER.isDebugEnabled()){
                    LOGGER.debug("Received topic: {}, message: {}", topic, receivedMessage);
                }
                for (Pair<LocalFilter, MessageHandler> pair : mArgsMap.values()){
                    if(pair.first.test(receivedMessage)) {
                        pair.second.onMessage(mSocket, receivedMessage);
                    }
                }
            }
        });
    }

    public long addHandler(LocalFilter filter, MessageHandler handler){
        long id = mRegSerialId.nextId();
        mArgsMap.put(id, Pair.of(filter, handler));
        return id;
    }

    public void removeHandler(long id){
        mArgsMap.remove(id);
    }
}
